package com.yfbao.horizon.schedule.tk.log;

import com.google.common.collect.Queues;
import lombok.extern.slf4j.Slf4j;
import org.osgi.service.log.LogEntry;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@Slf4j
public class LogDispatcher {
    // 内存队列配置（200G/日场景建议值）
    private static final int QUEUE_CAPACITY = 100_000;
    private final BlockingQueue<LogEntry> logQueue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);

    // 写入线程池
    private final ExecutorService writerExecutor = Executors.newFixedThreadPool(1);

    private volatile boolean running = true;
    private final AtomicInteger consecutiveEmpties = new AtomicInteger(0);
    private int batchSize = 500;

    @PostConstruct
    public void init() {
        // 启动4个写入线程（根据CPU核心数调整）
        for (int i = 0; i < 1; i++) {
            writerExecutor.submit(this::writeToMappedFile);
        }
    }

    public boolean addLog(LogEntry entry) {
        return logQueue.offer(entry);
    }

    public boolean addBatchLog(List<LogEntry> entrys) {
        return logQueue.addAll(entrys);
    }

    private void adjustBatchSize(){
        this.consecutiveEmpties.addAndGet(10);
    }

    private void writeToMappedFile() {
        long lastBusyTime = System.currentTimeMillis();
        int emptyCycles = 0;

        while (running) {
            try {
                // 背压检测（如果持续忙碌超过阈值，增加批量大小）
                if (System.currentTimeMillis() - lastBusyTime > 5000) {
//                    log.info("背压检测+");
                    adjustBatchSize();
                    lastBusyTime = System.currentTimeMillis();
                }
                List<LogEntry> batch = new ArrayList<>(batchSize);
                int drained = Queues.drain(logQueue, batch, batchSize, 100, TimeUnit.MILLISECONDS);

                if (drained > 0) {
//                    log.info("Drained {} logs", drained);
                    emptyCycles = 0;
                    lastBusyTime = System.currentTimeMillis();
                    MappedFileWriter.writeBatch(batch);
                } else {
                    emptyCycles++;
                    // 指数退避休眠
                    long sleepTime = Math.min(50 * (1L << Math.min(emptyCycles, 4)), 1000);
                    Thread.sleep(sleepTime);
                }
            } catch (InterruptedException e) {
                log.error("Log write InterruptedException error ", e);
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("Log write error", e);
                backoffSleep();
            }
        }
    }

    private void backoffSleep() {
        try {
            Thread.sleep(100 + (new Random()).nextInt(100));
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
